package com.alex;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.TimeUnit;

/**
 * 使用DefaultMQProducer发送消息
 *
 * @author shufei
 */
public class DefaultMQProducerTest {

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("MUKE_GROUP");
        producer.setInstanceName("localhost");
        producer.setRetryTimesWhenSendFailed(3);
        producer.setNamesrvAddr("192.168.1.104:9876");
        producer.start();

        for (int i = 0; i < 1000; i++) {
            Message message = new Message("TopicTest",
                    "TagA",
                    ("Hello RocketMq" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

            // 1s/5s/10s/30s/1m/2m/3m/4m/5m/6m/7m/8m/9m/10m/20m/30m/1h/2h）。比如setDelayTimeLevel(3)表示延迟10s
            message.setDelayTimeLevel(3);

            try {
                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println(sendResult);
                        System.out.println("status: " + sendResult.getSendStatus());
                        /**
                         * sendStatus:
                         *  1. SEND_OK: 表示发送成功。没有发生其它三种情况，即认为是成功
                         *  2. FLUSH_DISK_TIMEOUT: 表示没有在规定的时间内完成刷盘(需要Broker的刷盘策略呗设置为SYNC_FLUSH才会报这个错误)
                         *  3. FLUSH_SLAVE_TIMEOUT: 表示在主备方式下，并且Broker被设置为了SYNC_MASTER方式，且没有在设定时间内完成主从同步
                         *  4. SLAVE_NOT_AVAILABLE: 同FLUSH_SLAVE_TIMEOUT，但是，不同的是没有找到被配置为salve的Broker
                         */
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        throwable.printStackTrace();
                    }
                });
            } catch (InterruptedException e) {
                e.printStackTrace();
                TimeUnit.MINUTES.sleep(1000);
            }
        }

        producer.shutdown();
    }
}
